package com.lzwk.tableSql.kafka.source;

import com.lzwk.utils.MyKafkaUtil;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: CC
 * @Date: 2022/2/18 14:10
 * nginx-ingress-log-->ods_earth_log_clean,dwd_wk_log,dwd_wks_log,dwd_wka_log,dwd_lha_log,dwd_oc_log,dwd_lx_log,dwd_others_log,dirty_earth_log
 */
public class EarthLog {
    public static void odsEarthLog(StreamTableEnvironment tableEnv, String topic, String groupId) {
        String sql = "create table " + topic + "(" +
                "            `pre_spm` STRING," +
                "            `ref` STRING," +
                "            `ts` TIMESTAMP(3)," +
                "            `ua` STRING," +
                "            `url` STRING," +
                "            `url_params` MAP<STRING, STRING>," +
                "            `spm` STRING," +
                "            `tor_id` STRING," +
                "            `sid` STRING," +
                "            `uid` STRING," +
                "            `rts` STRING," +
                "            `oid` STRING," +
                "            `evt` STRING," +
                "            `evt_params` MAP<STRING, STRING>," +
                "            `ip` STRING," +
                "            `tsk` STRING," +
                "            `dt` STRING," +
                "            `hr` STRING," +
                "         WATERMARK FOR `ts` AS `ts` - INTERVAL '90' SECOND" +
                ") with (" + MyKafkaUtil.getKafkaDDL(topic, groupId) + ")";
        System.out.println("odsEarthLog>>>>>>"+sql);
        tableEnv.executeSql(sql);
    }

    /**
     * 统一格式流写入
     * dwd_wk_log
     * dwd_wks_log
     * dwd_wka_log
     * dwd_lha_log
     * dwd_oc_log
     * dwd_lx_log
     * dwd_others_log
     */
    public static void dwdEarthLog(StreamTableEnvironment tableEnv, String topic, String groupId) {
        String sql = "create table " + topic + "(" +
                "        `uid` STRING," +
                "        `spm` STRING," +
                "        `evt` STRING," +
                "        `evt_params` MAP<STRING, STRING>," +
                "        `ts` TIMESTAMP(3)," +
                "        `rts` STRING," +
                "        `sid` STRING," +
                "        `tor_id` STRING," +
                "        `tsk` STRING," +
                "        `url_pat` STRING," +
                "        `url` STRING," +
                "        `url_params` MAP<STRING, STRING>," +
                "        `spm0` STRING," +
                "        `spm1` STRING," +
                "        `spm2` STRING," +
                "        `spm3` STRING," +
                "        `dt` STRING," +
                "        `hr` VARCHAR," +
                "        WATERMARK FOR `ts` AS `ts` - INTERVAL '90' SECOND" +
                ") with (" + MyKafkaUtil.getKafkaDDL(topic, groupId) + ")";
        System.out.println("dwdEarthLog>>>>>>"+sql);
        tableEnv.executeSql(sql);
    }

    public static void dirtyEarthLog(StreamTableEnvironment tableEnv,  String topic, String groupId) {
        String sql = "create table " + topic + "(" +
                "            `@timestamp` VARCHAR," +
                "            `@filepath` VARCHAR," +
                "            `log` VARCHAR" +
                ") with (" + MyKafkaUtil.getKafkaDDL(topic, groupId) + ")";
        System.out.println("dirtyEarthLog>>>>>>"+sql);
        tableEnv.executeSql(sql);
    }

}

